/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */

package cgl.narada.protocol;

import java.util.Hashtable;
import java.util.LinkedList;

import cgl.narada.event.EventHeaders;
import cgl.narada.event.EventID;
import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.service.ServiceException;
import cgl.narada.service.client.ClientService;
import cgl.narada.service.client.EventProducer;
import cgl.narada.service.client.SessionService;

/**
 * This thread keeps track of set of event identifiers. The event identifiers
 * are purged after a certain interval (ID_LIFECYCLE) has elapsed. The oldest
 * set of entries are also purged if the number of entries in the buffers has
 * exceeded the maximum specified limit (MAX_LENGTH_OF_ID_BUFFER).
 * 
 * @author Shrideep Pallickara
 * 
 */

public class DuplicateDetection extends Thread implements ProtocolDebugFlags {

    private final long TIME_TO_SLEEP = 500;

    private final long ID_LIFECYCLE = 3000;

    private int MAX_LENGTH_OF_ID_BUFFER = 10000;

    private boolean keepLooping = true;

    private Hashtable<String, DDEntry> entries;

    private LinkedList<DDEntry> entriesList;

    private Object syncObject;

    private long tracker = 0;

    private String moduleName = "DuplicateDetection: ";

    private static final DuplicateDetection instance = new DuplicateDetection();

    private DuplicateDetection() {
        entries = new Hashtable<String, DDEntry>();
        entriesList = new LinkedList<DDEntry>();
        syncObject = new Object();
        this.setDaemon(true);
        start();
    }

    public static synchronized DuplicateDetection getInstance() {
        return instance;
    }

    /** Shutdown the thread's services */
    public void shutdownServices() {
        keepLooping = false;
    }

    /**
     * Restart the thread's services.
     * 
     */
    public void restartServices() {
        keepLooping = true;
        start();
    }

    /**
     * Check to see if an EventID is duplicate one.
     * 
     * @param nbEvent
     * @return true if this is a duplicate event, false otherwise.
     */
    public boolean isDuplicate(NBEvent nbEvent) {
        boolean isDuplicate = false;

        EventHeaders headers = nbEvent.getEventHeaders();
        if (!headers.hasEventId()) {
            return isDuplicate;
        }

        EventID eventId = headers.getEventId();
        int clientId = eventId.getEntityID();
        int sequenceNumber = eventId.getSequenceNumber();
        long timestamp = eventId.getTimeStamp();

        String eventIdentifier = clientId + "/" + sequenceNumber + "/"
            + timestamp;

        if (entries.containsKey(eventIdentifier)) {
            isDuplicate = true;
            return isDuplicate;
        }

        if (DuplicateDetection_Debug) {
            System.out.println(moduleName + "Entries.size() = "
                + entries.size() + ", list.size() = " + entriesList.size());
        }
        return isDuplicate;
    }

    /**
     * Once an event has been deemed to be not a duplicate, such an event needs
     * to be registered with the duplicate detector so that future duplicates
     * can be detected.
     * 
     * @param nbEvent
     */
    public void registerEvent(NBEvent nbEvent) {
        EventHeaders headers = nbEvent.getEventHeaders();
        if (!headers.hasEventId()) {
            return;
        }

        EventID eventId = headers.getEventId();
        int clientId = eventId.getEntityID();
        int sequenceNumber = eventId.getSequenceNumber();
        long timestamp = eventId.getTimeStamp();

        DDEntry ddEntry = new DDEntry(clientId, timestamp, sequenceNumber);

        String eventIdentifier = ddEntry.getIdentifier();

        synchronized (syncObject) {
            entries.put(eventIdentifier, ddEntry);
            entriesList.add(ddEntry);
        }
    }

    /**
     * This method checks the length of the buffers to see if the lengths have
     * been exceeded. If so, the excess IDs are purged in an FIFO fashion.
     */
    private void performBufferLengthCheck() {
        int bufferSize = entries.size();
        if (DuplicateDetection_Debug) {
            System.out.println(moduleName + "Buffer Size =" + bufferSize);
        }
        if (bufferSize <= MAX_LENGTH_OF_ID_BUFFER) {
            /**
             * There are no entries that need to be purged. Simply proceed to
             * the next step.
             */
            return;
        }

        int numOfEntriesToBePurged = bufferSize - MAX_LENGTH_OF_ID_BUFFER;

        for (int i = 0; i < numOfEntriesToBePurged; i++) {
            removeFirstEntry();
        }

        if (DuplicateDetection_Debug) {
            System.out.println(moduleName + "Purged (" + numOfEntriesToBePurged
                + ") entries based on buffer length check.");
        }

    }

    /**
     * Cycle through the entries in the linked list. Check to see if any of the
     * entries have decayed. If they have decayed, proceed to remove them from
     * the list. Since new entries are added to the bottom of the list, if the
     * head of the list has not decayed, none of the other entries in the list
     * have decayed either.
     */
    private void purgeDecayedEntries() {
        /**
         * Cycle through the IDs and see which of these IDs have elapsed. If the
         * ID has expired proceed to remove it from the list of IDs being
         * maintained
         */
        boolean keepProcessing = true;

        long currentTime = System.currentTimeMillis();
        int decayedEntries = 0;
        while (keepProcessing) {
            if (entriesList.size() == 0) {
                return;
            }
            DDEntry ddEntry = entriesList.getFirst();
            long entryTimestamp = ddEntry.getEntryTimestamp();

            long entryLifetime = currentTime - entryTimestamp;

            if (entryLifetime < ID_LIFECYCLE) {
                /**
                 * Since the first entry of the list has not yet decayed, none
                 * of the subsequent entries would have decayed either. Simply
                 * exit.
                 */
                keepProcessing = false;
                continue;
            }
            decayedEntries++;
            removeFirstEntry();
        }

        if (DuplicateDetection_Debug) {
            System.out.println(moduleName + "Decayed Entries in this pass ="
                + decayedEntries);
        }
    }

    /**
     * Removes the first entry from the linked list, and also the corresponding
     * entry from the hashtable.
     */
    private void removeFirstEntry() {
        synchronized (syncObject) {
            tracker++;
            DDEntry ddEntry = entriesList.removeFirst();
            removeEntry(ddEntry);
        }
    }

    /**
     * Removes a specific entry from the ID buffers.
     * 
     * @param ddEntry
     */
    private void removeEntry(DDEntry ddEntry) {
        String eventIdentifier = ddEntry.getIdentifier();

        if (entries.containsKey(eventIdentifier)) {
            entries.remove(eventIdentifier);
        } else {
            String errorReport = moduleName
                + "Unknown event-identifier submitted for removal";
            System.out.println(errorReport);
        }
    }

    public void run() {
        while (keepLooping) {
            try {
                Thread.sleep(TIME_TO_SLEEP);

                performBufferLengthCheck();
                purgeDecayedEntries();
                if (DuplicateDetection_Debug) {
                    System.out.println(moduleName + "Entries purged ="
                        + tracker);
                }
            } catch (InterruptedException e) {
                System.out.println(moduleName + "Problems sleeping!");
                e.printStackTrace();
            }
        }
    }

    /**
     * This inner class holds the entry that is maintained by the duplicate
     * detector thread.
     * 
     * @author Shrideep Pallickara
     * 
     */
    private class DDEntry {

        private int clientId;

        private long timestamp;

        private int sequenceNumber;

        private long entryTimestamp;

        private String identifier;

        public DDEntry(int clientId, long timestamp, int sequenceNumber) {
            this.clientId = clientId;
            this.timestamp = timestamp;
            this.sequenceNumber = sequenceNumber;
            entryTimestamp = System.currentTimeMillis();
            identifier = clientId + "/" + sequenceNumber + "/" + timestamp;
        }

        /**
         * This is the timestamp at which the entry was created
         * 
         * @return
         */
        public long getEntryTimestamp() {
            return entryTimestamp;
        }

        /**
         * The identifier created for this entry
         * 
         * @return
         */
        public String getIdentifier() {
            return identifier;
        }
    }

    /**
     * @param args
     */
    public static void main(String[] args) {
        int entityId = 7007;
        DuplicateDetection ddThread = DuplicateDetection.getInstance();

        try {
            ClientService clientService = SessionService
                .getClientService(entityId);
            EventProducer producer = clientService.createEventProducer();
            producer.generateEventIdentifier(true);

            byte[] testBytes = "Test for Detection".getBytes();
            String topic = "/Movies/Casablanca";

            for (int i = 0; i < 20000; i++) {
                NBEvent nbEvent = producer
                    .generateEvent(TemplateProfileAndSynopsisTypes.STRING,
                                   topic, testBytes);

                boolean isDuplicate = ddThread.isDuplicate(nbEvent);
                if (!isDuplicate) {
                    ddThread.registerEvent(nbEvent);
                }

                if (i % 100 == 0) {
                    System.out.println("Sleeping at iteration (" + i + ")");
                    Thread.sleep(10);
                }
            }

        } catch (ServiceException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

}
